package yuekao2.util;

import com.alibaba.fastjson.JSON;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class MyHbase extends RichSinkFunction<String> {
    Connection conn=null;
    PreparedStatement ps=null;
    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
        conn=DriverManager.getConnection("jdbc:phoenix:hadoop-single:2181","","");
        conn.setAutoCommit(true);
    }

    @Override
    public void close() throws Exception {
        conn.close();
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        String table = "dim_" + JSON.parseObject(value).getString("table");
        Integer id = JSON.parseObject(value).getJSONObject("data").getInteger("id");
        if(id==null&& id.equals("")){
            id = 1;
        }
        ps = conn.prepareStatement("upsert into "  + table + " values(?,?)");
        ps.setObject(1, id);
        ps.setObject(2, value);
        ps.execute();
        ps.close();
    }
}
